package highlevel;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.*;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Produced;

import java.util.ArrayList;
import java.util.Properties;

/**
 * @Author:RenPu
 * @Date: 2020/4/17 10:41
 * @Version: 1.0
 * @description:kafka streaming highlevel 高级水平api单词计数的使用
 */
public class HighLevelWordCount {


        public static void main(String[] args) {
          //创建properties对象配置kafka集群以及相关
            Properties properties = new Properties();
            properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"node1:9092,node2:9092,node3:9092");
            properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
            properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG,Serdes.String().getClass());
            properties.put(StreamsConfig.APPLICATION_ID_CONFIG,"word-dsl");
            properties.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG,3);
            

            //创建StreamBuild对象，可以获取kafkastreaming的topology任务
            StreamsBuilder streamsBuilder = new StreamsBuilder();

            
            //从消费主题进行获取record中的数据
            KStream<String, String> kStream= streamsBuilder.stream("t9");
            
            //数据分组，计算，以及数据组织
            KTable<String, Long> kTable = kStream.flatMap((k, v) -> {
                String[] word = k.split(" ");
                ArrayList<KeyValue<String, String>> list = new ArrayList<>();

                for (int i = 0; i < word.length; i++) {
                    KeyValue<String, String> keyValue = new KeyValue<>(k, v);
                    list.add(keyValue);
                }
                return list;

            }).groupBy((k, v) -> v).count();


            //将计算的结果写入指定的主题
            kTable.toStream().to("t10", Produced.with(Serdes.String(),Serdes.Long()));

            //获取topology任务对象
            Topology topology = streamsBuilder.build();

            //创建kafkaStream对象启动任务
            KafkaStreams kafkaStreams = new KafkaStreams(topology,properties);

            //开启任务
            kafkaStreams.start();


        }





}
